package com.bw.yk01;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.List;

public class JsonDebeziumDeserializationSchema implements DebeziumDeserializationSchema<String> {

    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {

        JSONObject all = new JSONObject();

        String[] split = sourceRecord.topic().split("\\.");
        String db = split[1];//获取数据库名称
        String tb = split[2];//获取表名
        all.put("db",db);
        all.put("tb",tb);

        Struct value = (Struct) sourceRecord.value();
        String op = value.getString("op"); //获取操作类型
        all.put("op",op);

        Struct after = value.getStruct("after"); //获取数据，如果是删除则没有after
        JSONObject data = new JSONObject();
        if(after!=null){
            //after.getString("字段名")
            //遍历after中的字段
            List<Field> fields = after.schema().fields();
            for(Field f: fields){
                String c = f.name();//字段名称
                Object v = after.get(c);
                v = checkType(c, v); // 金额转换和日期转换
                data.put(c,v);
            }
        }
        all.put("data",data);

        collector.collect(all.toString());

    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }

    public static Object checkType(String c,Object v) {

        if(c.endsWith("_time") && v!=null && v instanceof Long){
            Timestamp dateTimeValue =new Timestamp((Long) v);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            v = sdf.format(dateTimeValue);
        }

        if(c.endsWith("_amount") && v!=null){
            BigDecimal decimalValue = (BigDecimal)v;
            v=decimalValue.doubleValue();
        }
        return v;
    }
}
